package com.hadooop.topN;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Comparator;
import java.util.PriorityQueue;

/*
这里是修改前的reduce类
这里的优先队列里放的是IntWritable
这样会出现一个问题，就是每add一个新的元素都会把之前的元素也变成这个新的元素。
原因可能是：
每次运行reduce方法时传入的key都是同一个IntWritable对象。
在同一个reduceTask中reduce方法中的key会重用，每次新运行一次，都只改变了原来的key的值，而没有重新创建一个新的IntWritable对象
 */
public class TopNReducer extends Reducer<IntWritable, NullWritable, NullWritable, IntWritable> {

    private int N = 5;

    private PriorityQueue<IntWritable> topNQueue = new PriorityQueue<IntWritable>(N+1, new Comparator<IntWritable>() {
        @Override
        public int compare(IntWritable o1, IntWritable o2) {
            return o1.get() - o2.get();
        }
    });

    @Override
    protected void reduce(IntWritable key, Iterable<NullWritable> values, Reducer<IntWritable, NullWritable, NullWritable, IntWritable>.Context context) throws IOException, InterruptedException {
        topNQueue.add(key);
        if (topNQueue.size() > N) {
            topNQueue.poll();
        }
    }

    @Override
    protected void cleanup(Reducer<IntWritable, NullWritable, NullWritable, IntWritable>.Context context) throws IOException, InterruptedException {
        while (!topNQueue.isEmpty()) {
            context.write(NullWritable.get(), topNQueue.poll());
        }
    }
}
